package com.study.flink.sink

import java.sql.Connection
import java.util.Properties

import com.alibaba.druid.pool.DruidDataSourceFactory
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
  * 自定义Sink
  * JDBC Sink Demo
  *
  * @author stephen
  * @date 2019-07-22 15:26
  */
object ScalaJDBCSinkDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取数据
    import org.apache.flink.api.scala._
    val dataStream = env.socketTextStream("localhost", 9999)

    // 3 word count
    val wordAndCountStream = dataStream.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1).map(data=>Array(data._1,data._2))

    // 4 输出到mysql
    val jdbcSink = new JDBCSink("REPLACE INTO wc(word,count) VALUES(?, ?);")
    wordAndCountStream.addSink(jdbcSink)

    // 5 启动任务
    env.execute("JDBC Sink Demo")
  }
}

class JDBCSink(sql: String) extends RichSinkFunction[Array[Any]] {

  val driver = "com.mysql.jdbc.Driver"
  val url = "jdbc:mysql://192.168.8.36:3306/test?useSSL=false"
  val username = "root"
  val password = "123456"
  val maxActive = "20"
  var connection: Connection = null;

  override def open(parameters: Configuration): Unit = {
    // 使用druid
    val properties = new Properties()
    properties.put("driverClassName", driver)
    properties.put("url", url)
    properties.put("username", username)
    properties.put("password", password)
    properties.put("maxActive", maxActive)

    val dataSource = DruidDataSourceFactory.createDataSource(properties)
    connection = dataSource.getConnection
    connection.setAutoCommit(false)

    // 不使用druid
    //Class.forName("com.mysql.jdbc.Driver")
    //connection = DriverManager.getConnection(url, username, password)
    //connection.setAutoCommit(false)

  }

  override def invoke(value: Array[Any], context: SinkFunction.Context[_]): Unit = {
    // 预编译
    val ps = connection.prepareStatement(sql)
    for (i <- 0 until value.length) {
      // 坐标是从1开始的
      ps.setObject(i + 1, value(i))
    }
    // 执行操作
    ps.executeUpdate()
    connection.commit()

  }

  override def close(): Unit = {
    if (connection != null) {
      connection.close()
    }
  }
}